package cn.itcast.streaming.source;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * Author itcast
 * Date 2021/9/26 16:25
 * Desc TODO
 */
public class MysqlElectricFenceResultSource extends RichSourceFunction<HashMap<String,Integer>> {
    private static final Logger logger = LoggerFactory.getLogger(MysqlElectricFenceResultSource.class.getSimpleName());
    private static Connection conn = null;
    private static Statement statement = null;
    //设置标识用于记录当前循环读取mysql配置
    private static Boolean flag = true;
    private static ParameterTool globalJobParameters = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        //1.1 获取上下文中的 parameterTool
        globalJobParameters = (ParameterTool) getRuntimeContext()
                .getExecutionConfig().getGlobalJobParameters();
        //1.2 读取配置文件中，注册驱动 url user password
        String driver = globalJobParameters.getRequired("jdbc.driver");
        String url = globalJobParameters.getRequired("jdbc.url");
        String user = globalJobParameters.getRequired("jdbc.user");
        String password = globalJobParameters.getRequired("jdbc.password");
        //设置驱动和连接
        Class.forName(driver);
        conn = DriverManager.getConnection(url, user, password);
        //1.3 实例化statement
        statement = conn.createStatement();
    }

    @Override
    public void run(SourceContext<HashMap<String,Integer>> ctx) throws Exception {
        while (flag) {
            HashMap<String,Integer> vehInfoMap = new HashMap<String,Integer>();
            String sql = "select vin, min(id) id from vehicle_networking.electric_fence where inTime is not null and outTime is null GROUP BY vin";
            ResultSet rs = statement.executeQuery(sql);
            while(rs.next()){
                vehInfoMap.put(rs.getString("vin"), rs.getInt("id"));
            }
            ctx.collect(vehInfoMap);
            TimeUnit.MILLISECONDS.sleep(globalJobParameters.getLong("vehinfo.millionseconds"));
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() throws Exception {
        //2.1 关闭 statement 和 conn
        if (!statement.isClosed()) statement.close();
        if (!conn.isClosed()) conn.close();
    }
}
